package com.atguigu.flink.datastramapi.source;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

/**
 * Created by Smexy on 2023/2/24
 *
 *  Connector： 连接器。
 *              使用flink程序连接外部系统(数据库，文件系统，中间件)
 *
 *     读取kafka，必须先引入和kafka相关的连接器。
 *
 *     读取kafka，flink程序需要使用一个消费者 FlinkKafkaConsumer。
 *
 *     1.13未过时。
 *     1.14以后，将过时。
 *
 *   ------------
 *      本质上创建就是一个消费者组，这个组中起N(parilisim)个消费者
 */
public class Demo2_KafkaSource
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        /*
            创建 FlinkKafkaConsumer

            FlinkKafkaConsumer(
            String topic, 消费的主题
            DeserializationSchema<T>   valueDeserializer,： 反序列化器。
                                    用于把value反序列化为指定的类型。
            Properties props： 消费者参数设置。必须要设置  集群地址，组id
                        所有的消费者参数，参考 ConsumerConfig
            )

            默认自动将消费的offsets信息，提交到kafka的 _consumer_offsets主题中
         */
        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"atguigu");
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");

        FlinkKafkaConsumer flinkKafkaConsumer = new FlinkKafkaConsumer("topicA", new SimpleStringSchema(), properties);

        //使用消费者读取kafka
        env.addSource(flinkKafkaConsumer)
           .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
